package Autodesk;


import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;

public class LookForNotFound extends Configured implements Tool  {
	
		public Path prodPath;
		public static class LogMapper extends Mapper<Object, Text, Text, IntWritable> {
			private static final String EXCLUDE_VALUES = "beehive_monitor";  
			private static final String EXCLUDE_STYLES = "/styles/";  
			private static final String EXCLUDE_JS = ".js";  
			private static Logger logger = Logger.getLogger(ParseLogs.class);


			private final static IntWritable one = new IntWritable(1);
			private Text word = new Text();
			private Map<ArrayList<String>,String> StaticListLang=new HashMap<ArrayList<String>,String>();
			private Map<ArrayList<String>,String> StaticListProd=new HashMap<ArrayList<String>,String>();
			private Map<ArrayList<String>,String> StaticListChannel=new HashMap<ArrayList<String>,String>();
			private Map<ArrayList<String>,String> StaticListFeature=new HashMap<ArrayList<String>,String>();
			private ArrayList<String> listChannelsExcluded=new ArrayList<String>();
			Parser par=new Parser();
			
			public void setup(Context context){
				Configuration conf=context.getConfiguration();
//				 if (conf.getBoolean("language.patterns", false) | conf.getBoolean("product.patterns", false) | conf.getBoolean("feature.patterns", false) |
//						 conf.getBoolean("channel.patterns", false) | conf.getBoolean("version.patterns", false) | conf.getBoolean("time.patterns", false) ){
//					  Path[] patternsFiles = new Path[0];
//					  try{
//				           patternsFiles = DistributedCache.getLocalCacheFiles(conf);
//					  }catch(IOException ioe){
//				           System.err.println("Caught exception while getting cached files: ");
//					  }
//					  for (Path patternsFile : patternsFiles) {
//						  
//						  	if( (conf.get("language.path")).equals(patternsFile.getName())){
//						  		par.getStaticList(patternsFile,StaticListLang,"lang"); 
//						  	}else if((conf.get("product.path").equals(patternsFile.getName()))){
//						  		par.getStaticList(patternsFile,StaticListProd,"products"); 
////						  	}else if((conf.get("feature.path").equals(patternsFile.getName()))){
////						  		par.getStaticList(patternsFile,StaticListFeature,"features"); 
//						  	}else if((conf.get("channel.path").equals(patternsFile.getName()))){
//						  		par.getStaticList(patternsFile,StaticListChannel,"channel"); 
//						  	}
//					  }
//					 
//				 }
 				StaticListLang=par.getStaticList("language"); 
 				logger.info("Loaded Langs-------------------");
		  		
		  		StaticListProd=par.getStaticList("product"); 
		  		logger.info("Loaded Products-------------------");
		  		
		  		StaticListChannel=par.getStaticList("channel"); 
		  		logger.info("Loaded Channels-------------------");
		  		
				 inicitializeListChannelsExc();

			}
			private void inicitializeListChannelsExc() {
				listChannelsExcluded.add("productchannel");
				 listChannelsExcluded.add("productchannel2");
				 listChannelsExcluded.add("supportchannel");
				 listChannelsExcluded.add("supportchannel2");
				 listChannelsExcluded.add("acadconnectchannel");
				 listChannelsExcluded.add("acadconnectchannel2");
				 listChannelsExcluded.add("taxonomy3");
			}
				
		private String generateKeyProducts(String queryline,String tokenProd, String tokenRele) {
			String tokenSub;
			if(queryline.indexOf("origin=upi")<0 ){//IT IS A BASELINE VALUE
				tokenSub=tokenProd+"\t"+""+"\t"+tokenRele;
			}else{
				tokenSub=""+"\t"+tokenProd+"\t"+tokenRele;
			}
			return tokenSub;
		}
		
		
		//MAP FUNCTION
		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

			//String line = value.toString().toLowerCase();
			String line = value.toString();
			String channel = null;

			// Looking for CHANNEL.
			String channelID;
			channel = par.getChannel(line);

			channelID= par.lookForKey(StaticListChannel, channel.toLowerCase(),"channel");
			if ( (channelID.equals("1")) && !(listChannelsExcluded.contains(channel))){
				word.set("channels"+"\t"+channel);
			}
			if( (line.indexOf(EXCLUDE_STYLES) > -1) && (line.indexOf(EXCLUDE_JS)> -1) && listChannelsExcluded.contains(channel) && line.indexOf(EXCLUDE_VALUES)>-1){				
				// skip
			} else if(line.indexOf("?")>-1){
				String queryLine = par.getQueryLine(line);
					StringTokenizer tokenizer = new StringTokenizer(queryLine,"&");
					while (tokenizer.hasMoreTokens()) {
						String token = tokenizer.nextToken();							
							if (token.indexOf("p=") > -1) {
								try{
									String prodAb=par.getProduct(token, queryLine,par);								
									String[] productRele=prodAb.split("\t");
									String tokenProd=productRele[0];
									String tokenRele=productRele[1];
									String id=par.lookForKeyProduct(StaticListProd, tokenProd.toLowerCase(),tokenRele.toLowerCase());
									if(id.equals("1") && !tokenProd.equals("null")){
										id=generateKeyProducts(queryLine, tokenProd, tokenRele);
										word.set("products"+"\t"+id);
									}
								}catch(Exception e){
								}
							} else if ((token.indexOf("l=") > -1) && (token.indexOf("channel=") < 0) && (token.indexOf("url=") < 0)) {

								String tokenSub = token.substring(2);
								if(par.lookForKey(StaticListLang, tokenSub.toLowerCase(),"languages").equals("1")){
									word.set("languages"+"\t"+tokenSub);
								}								
//							} else if (token.indexOf("source=") > -1 || token.indexOf("audience=")>-1
//									|| token.indexOf("sc=")>-1 || token.indexOf("c=")>-1 || token.indexOf("cg=")>-1 || token.indexOf("origin=")>-1) {
//								String featValue = token.substring(token.indexOf("=")+1);
//								String featCategory=token.substring(0,token.indexOf("="));
//								if(par.lookForKey(StaticListFeature, featValue.toLowerCase(),"features").equals("1") & !featValue.equals("")){
//									word.set("features"+"\t"+featCategory+"\t"+featValue);
//								}
							}
						
						context.write(word, one);
				}
			}

		}

	}

	//REDUCE FUNCTION	
	public static class LogReducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		
		private IntWritable result = new IntWritable();
		 private MultipleOutputs<Text,IntWritable> mos;
		 public void setup(Context context) {
		 mos = new MultipleOutputs<Text,IntWritable>(context);
		 }
		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			result.set(sum);
			String[] recordIdentifier=key.toString().split("\t",2);
			if(recordIdentifier[0].equals("products")){
				mos.write("ProductsNotFound", new Text(recordIdentifier[1]), result);
			}else if(recordIdentifier[0].equals("languages")){
				mos.write("LanguagesNotFound", new Text(recordIdentifier[1]), result);
			}else if(recordIdentifier[0].equals("channels")){
				mos.write("ChannelsNotFound", new Text(recordIdentifier[1]), result);
//			}else if(recordIdentifier[0].equals("features")){
//				mos.write("FeaturesNotFound", new Text(recordIdentifier[1]), result);
			}else{
				System.out.println("ERROR: "+key.toString());
			}
			context.write(key, result);
		}
		 public void cleanup(Context context) throws IOException, InterruptedException {
			 mos.close();
			 }
	}

	public int run(String[] args) throws Exception { 
		Configuration conf = getConf();
		
		Job job = new Job(conf, "ParseLogs");
		job.setJarByClass(Autodesk.LookForNotFound.class);
		
//		FileInputFormat.addInputPath(job, new Path(args[0]));
//		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		job.setMapperClass(LogMapper.class);
		job.setReducerClass(LogReducer.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		List<String> other_args = new ArrayList<String>();
		for (int i = 0; i < args.length; ++i) {
			other_args.add(args[i]);
		}
//		for (int i = 0; i < args.length; ++i) {
//			if ("-languages".equals(args[i])) {
//				prodPath=new Path(args[++i]);
//				DistributedCache.addCacheFile(prodPath.toUri(),
//						job.getConfiguration());
//				job.getConfiguration().setBoolean("language.patterns",
//						true);
//				job.getConfiguration().set("language.path", prodPath.getName());
//			} else if ("-products".equals(args[i])) {
//				prodPath=new Path(args[++i]);
//				DistributedCache.addCacheFile(prodPath.toUri(),
//						job.getConfiguration());
//				job.getConfiguration().setBoolean("product.patterns",
//						true);
//				job.getConfiguration().set("product.path", prodPath.getName());
//			} else if ("-channels".equals(args[i])) {
//				prodPath=new Path(args[++i]);
//				DistributedCache.addCacheFile(prodPath.toUri(),
//						job.getConfiguration());
//				job.getConfiguration().setBoolean("channel.patterns",
//						true);
//				job.getConfiguration().set("channel.path", prodPath.getName());
//			} else if ("-features".equals(args[i])) {
//				prodPath=new Path(args[++i]);
//				DistributedCache.addCacheFile(prodPath.toUri(),
//						job.getConfiguration());
//				job.getConfiguration().setBoolean("feature.patterns",
//						true);
//				job.getConfiguration().set("feature.path", prodPath.getName());
//			}else{
//				other_args.add(args[i]);
//			}
//		}
	    FileInputFormat.setInputPaths(job, new Path(other_args.get(0)));
	    FileOutputFormat.setOutputPath(job, new Path(other_args.get(1)));
	    
	    MultipleOutputs.addNamedOutput(job, "ProductsNotFound", TextOutputFormat.class, Text.class, IntWritable.class);
	    MultipleOutputs.addNamedOutput(job, "LanguagesNotFound", TextOutputFormat.class, Text.class, IntWritable.class);
	    MultipleOutputs.addNamedOutput(job, "ChannelsNotFound", TextOutputFormat.class, Text.class, IntWritable.class);
	    MultipleOutputs.addNamedOutput(job, "FeaturesNotFound", TextOutputFormat.class, Text.class, IntWritable.class);


		System.exit(job.waitForCompletion(true)?0:1);
		
		return 0;
	
	}
	
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new LookForNotFound(), args);
		System.exit(res);
		}

 
}